package com.admin.config;

/*
import com.admin.service.ipl.WemediaNewsAutoScanService;
import com.usian.common.config.NewsAutoScanConstants;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class WemediaNewsAutoListener {
    @Autowired
    WemediaNewsAutoScanService wemediaNewsAutoScanService;
    //kakfa监听的主题
    @KafkaListener(topics = NewsAutoScanConstants.WM_NEWS_AUTO_SCAN_TOPIC)
    public void recevieMessage(ConsumerRecord<?,?> record) throws Exception {
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            //调用自动审核的方法  监听到主键id后  进行自动审核
            System.out.println("---------------------监听中---------------------");
            wemediaNewsAutoScanService.autoScanBymedianewsid(Integer.valueOf((String) value));
        }

    }
}*/
import com.admin.fen.ApFollowBehaviorfegin;
import com.admin.fen.Articlesfen;
import com.admin.service.ipl.WemediaNewsAutoScanService;
import com.alibaba.fastjson.JSON;

import com.usian.common.config.NewsAutoScanConstants;
import com.usian.model.behavior.FollowBehaviorConstants;
import com.usian.model.behavior.FollowBehaviorDto;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.xml.crypto.Data;
import java.util.HashMap;
import java.util.Optional;

@Component
public class WemediaNewsAutoListener {

    @Autowired
    WemediaNewsAutoScanService wemediaNewsAutoScanService;

    @Resource
    private Articlesfen articlesfen;

//    点赞功能
    @Resource
    private ApFollowBehaviorfegin apFollowBehaviorfegin;

    //kakfa监听的主题
    @KafkaListener(topics = NewsAutoScanConstants.WM_NEWS_AUTO_SCAN_TOPIC)
    public void recevieMessage(ConsumerRecord<?,?> record) throws Exception {
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            //调用自动审核的方法
            wemediaNewsAutoScanService.autoScanBymedianewsid(Integer.valueOf((String) value));
        }
    }



    //kakfa监听的主题  媒体上架修改
    @KafkaListener(topics = NewsAutoScanConstants.WM_NEWS_AUTO_ap_article_config)
    public void newsandconfig(ConsumerRecord<?,?> record) throws Exception {
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            //上架或者下架
            String s=value.toString();
            HashMap hashMap= JSON.parseObject(s,HashMap.class);
            System.out.println(hashMap);
            articlesfen.updateconfig(hashMap);
            System.out.println(" 媒体上架修改----------------------------------");

        }
    }

    //kakfa监听的主题  点赞列表的添加
    @KafkaListener(topics = FollowBehaviorConstants.FOLLOW_BEHAVIOR_TOPIC)
    public void behavior(ConsumerRecord<?,?> record) throws Exception {
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            //点赞  格式转换 将value转换成对应的 dto  用反射创建
            FollowBehaviorDto dto=JSON.parseObject(record.value().toString(),FollowBehaviorDto.class);
//          调用点赞列表中的service方法进行传递参数
            apFollowBehaviorfegin.saveFollowBehavior(dto);
            System.out.println(" 点赞----------------------------------");

        }
    }

}
